package asteroid

import (
	"fmt"
	"runtime/debug"
	"sync"

	"gddgame.cc/galaxy/utils"
	"gddgame.cc/galaxy/utils/def"
)

type PipeChannel interface {
	Pipe(key string) *Pipe
}

type Pub interface {
	Auto() *Pipe
	All() *Pipe
	Direct(node string) *Pipe
	Cross(info interface{}) *Pipe
}

type Publisher struct {
	def.Logger

	channel *channel
	key     string

	*responseManager
	*nodeManager
}

type Sub interface {
	On(event interface{}, f OnHandler)
}

type Subscriber struct {
	def.Logger

	channel *channel
	key     string
	on      map[interface{}]OnHandler
}

type PublishChannel interface {
	Publish(key string) (*Publisher, bool)
}
type SubscriberChannel interface {
	Subscribe(key string) *Subscriber
}
type PublishPipe interface {
	PipeChannel
	Publish(key string) (*Publisher, bool)
}
type Channel interface {
	PipeChannel
	PublishChannel
	SubscriberChannel
}
type channel struct {
	def.Logger

	layer *Layer
	node  string
	adapt Adapt

	publishers  sync.Map
	subscribers sync.Map

	mu sync.Mutex
}

func (channel *channel) setAdapt(adapt Adapt) error {
	channel.adapt = adapt
	return nil
}

func (channel *channel) dispatch(msg *Msg) bool {
	//msg.message.Logger = channel.LoggerWrap(channel.layer.trackKey, msg.Origin)
	switch msg.Type {
	case ResponseChannelMessageType:
		i, ok := channel.publishers.Load(msg.Key)
		if !ok {
			return false
		}
		msg.Alloc()
		publisher := i.(*Publisher)
		channel.layer.Thread(func() {
			//fmt.Println("publish", publisher.id, 1)
			publisher.onResponse(msg)
			//fmt.Println("publish", publisher.id, 2)
			msg.Release()
		})
	case RequestChannelMessageType:
		i, ok := channel.subscribers.Load(msg.Key)
		if !ok {
			return false
		}
		msg.Alloc()
		subscriber := i.(*Subscriber)
		msg.responseAction = subscriber
		channel.layer.Thread(func() {
			//fmt.Println("subscriber", msg.Key, 1)
			subscriber.onEvent(msg)
			//fmt.Println("subscriber", msg.Key, 2)
			msg.Release()
		})
		//fmt.Println("channel request finish")
	}
	return true
}
func (channel *channel) send(msg *Msg) {
	if msg.to == nil {
		msg.Release()
		return
	}
	//if msg.Key == "core" {
	//	fmt.Println("core", msg)
	//}
	msg.From = channel.node
	if err := msg.Send(ChannelMessage, nil); err != nil {
		channel.Error(err)
	}
}

func (channel *channel) Publish(key string) (*Publisher, bool) {
	if key, ok := channel.publishers.Load(key); ok {
		return key.(*Publisher), true
	}
	channel.mu.Lock()
	if key, ok := channel.publishers.Load(key); ok {
		channel.mu.Unlock()
		return key.(*Publisher), true
	}
	p := &Publisher{
		Logger:  channel.LoggerWrap("Publish", key),
		channel: channel,
		key:     key,
	}
	p.responseManager = newResponseManager(p, channel.layer.timer)
	p.nodeManager = newNodeManager(channel.layer, p, channel.node, false)
	channel.publishers.Store(key, p)
	channel.mu.Unlock()
	if err := channel.adapt.Monitor("publish:"+channel.node+key, key, p.join, p.leave, p.finish); err != nil {
		channel.Fatalf("Publish:%s Monitor %s", key, err)
	}
	channel.Debugf("Publish:%s", key)
	return p, false
}

func (channel *channel) Subscribe(key string) *Subscriber {
	if key, ok := channel.subscribers.Load(key); ok {
		return key.(*Subscriber)
	}
	channel.mu.Lock()
	if key, ok := channel.subscribers.Load(key); ok {
		channel.mu.Unlock()
		return key.(*Subscriber)
	}
	s := &Subscriber{
		Logger:  channel.LoggerWrap("Subscribe", key),
		channel: channel,
		key:     key,
		on:      make(map[interface{}]OnHandler),
	}
	channel.subscribers.Store(key, s)
	channel.mu.Unlock()
	if err := channel.adapt.Register("subscribe:"+channel.node+key, key, channel.layer.node); err != nil {
		channel.Fatalf("Subscribe:%s Register %s", key, err)
	}
	channel.Debugf("Subscribe:%s", key)
	return s
}

func (channel *channel) Pipe(key string) *Pipe {
	return channel.layer.Pipe().setSend(func(pipe *Pipe) {
		msg := newMsg(RequestChannelMessageType, "", key)
		msg.Pipe(pipe, nil)

		channel.send(msg)
	})
}
func (channel *channel) close() error {
	channel.subscribers.Range(func(key, value interface{}) bool {
		value.(*Subscriber).Close()
		return true
	})
	channel.publishers.Range(func(key, value interface{}) bool {
		value.(*Publisher).Close()
		return true
	})
	if channel.adapt != nil {
		if err := channel.adapt.Close(); err != nil {
			return err
		}
	}
	return nil
}

func (pub *Publisher) ID() string {
	return pub.key
}
func (pub *Publisher) sendPipe(pipe *Pipe) {
	cm := newMsg(RequestChannelMessageType, "", pub.key)
	cm.Pipe(pipe, pub.responseManager)
	pub.channel.send(cm)
}
func (pub *Publisher) Direct(node string) *Pipe {
	return pub.channel.layer.Pipe().setSend(pub.sendPipe).Node(node)
}
func (pub *Publisher) All() *Pipe {
	return pub.channel.layer.Pipe().setSend(pub.sendPipe).Nodes(pub.OtherNodes())
}
func (pub *Publisher) Auto() *Pipe {
	return pub.channel.layer.Pipe().setSend(func(pipe *Pipe) {
		node := pub.Alloc(pub.channel.node + utils.ToString(pipe.event))
		pipe.Node(node)
		pub.sendPipe(pipe)
	})
}
func (pub *Publisher) Cross(info interface{}) *Pipe {
	return pub.channel.layer.Pipe().setSend(pub.sendPipe).Cross(info)
}
func (pub *Publisher) Close() {
	_ = pub.channel.adapt.UnMonitor("publish:" + pub.channel.node + pub.key)
	pub.channel.publishers.Delete(pub.key)
}

func (sub *Subscriber) On(event interface{}, f OnHandler) {
	sub.on[event] = f
}
func (sub *Subscriber) onEvent(msg *Msg) {
	defer func() {
		if err := recover(); err != nil {
			debug.PrintStack()
			sub.Error(fmt.Errorf("Event error:%s,%s", msg.Event, err))
			sub.response(msg, err)
		}
	}()
	if handler, ok := sub.on[msg.Event]; ok {
		handler(msg.message)
	} else if handler, ok := sub.on[DefaultEvent]; ok {
		handler(msg.message)
	}
}
func (sub *Subscriber) response(req *Msg, data interface{}) {
	if req.Rid == 0 {
		if err, ok := data.(error); ok {
			sub.Error(err)
		}
		return
	}
	res := newMsg(ResponseChannelMessageType, "", sub.key)
	res.BindRequest(req, data)
	sub.channel.send(res)
}
func (sub *Subscriber) Close() {
	_ = sub.channel.adapt.UnRegister("subscribe:" + sub.channel.node + sub.key)
	sub.channel.subscribers.Delete(sub.key)
}
